package study.stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * spark stream 有状态实现准实时的wordcount
 * 为了解决多批次的运算
 *
 * @author zh
 * @date 2021/5/28 10:42
 */
object HasStateWordCount {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    // 第二个参数是采集周期
    val streamingContext = new StreamingContext(sparkConf, Seconds(3))

    // 设置检查点
    streamingContext.checkpoint("./checkpoint")

    val stream = streamingContext.receiverStream(new MyReceiver)

    val map = stream.map((_, 1))
    // updateStateByKey内需要传入一个函数
    //  函数第一个参数  values 为当前批次单词频度
    //  函数第二个参数  state 为以往批次单词频度
    val wordCount = map.updateStateByKey[Int]((values: Seq[Int], state: Option[Int]) => {
      val current = values.foldLeft(0)(_ + _)
      val pre = state.getOrElse(0)
      Some(current + pre)
    })

    wordCount.print()

    // 启动采集器
    streamingContext.start()

    // 等待采集器停止
    streamingContext.awaitTermination()

  }

}

